package com.advert.web;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.advert.common.core.controller.BaseController;
import com.advert.common.core.domain.AjaxResult;
import com.advert.common.core.redis.RedisCache;
import com.advert.common.utils.DateUtil;
import com.advert.common.utils.StringUtils;
import com.advert.entity.*;
import com.advert.mqtt.client.EmqClient;
import com.advert.mqtt.properties.MqttProperties;
import com.advert.system.domain.Devices;
import com.advert.system.service.IDevicesService;
import com.advert.system.util.RedisConst;
import com.advert.system.util.WXPayUtil;
import com.advert.utils.RequestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/api/device")
public class DeviceApiController extends BaseController {


    @Autowired
    private RequestUtils requestUtils;

    @Autowired
    private MqttProperties mqttProperties;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    EmqClient emqClient;

    @Autowired
    private IDevicesService devicesService;

    @RequestMapping(value = "/isOnline")
    public AjaxResult isOnline(String deviceId) {
        logger.info("mqttProperties====" + JSON.toJSONString(mqttProperties));
        ResponseEntity<MQTTClientStatusResp> clientResp = requestUtils.get(mqttProperties.getApiUrl() + "/api/v4/clients/GID_cabinet@@@" + deviceId);
        logger.info("clientResp====" + JSON.toJSONString(clientResp));

        if (clientResp.getBody().getData().size() == 0) {
            logger.error("设备号{} 请求获取在线接口错误", deviceId);
            return AjaxResult.error("请求获取在线接口错误");
        }
        return AjaxResult.success(true);
    }
    @PostMapping("/receiveMqtt")
    public AjaxResult receiveMqtt(@RequestBody JSONObject body) {
        long startTime = System.currentTimeMillis();
        // 设备号
        String deviceId = body.getString("deviceId");
//        JSONObject  = body.getJSONObject("bizContent");
        String event = body.getString("event");
        String key = body.getString("key");
        String biz  = body.getString("bizContent");
        if (emqClient == null) {
            return AjaxResult.error("emqClient不存在");
        }
        if (!emqClient.isConnected()) {
            logger.error("emqClient connect 不在线: {}", emqClient.isConnected());
            return AjaxResult.error("emqClient未连接");
        }
        // 查询目标客户端是否在集群中
        ResponseEntity<MQTTClientStatusResp> clientResp = requestUtils.get(mqttProperties.getApiUrl()  + "/api/v4/clients/GID_cabinet@@@" + deviceId);
        if (clientResp.getStatusCodeValue() != 200) {
            logger.error("设备号{} 请求获取在线接口错误", deviceId);
            return AjaxResult.error("请求获取在线接口错误");
        }
        MQTTClientStatusResp respData = clientResp.getBody();
        if (respData == null || respData.getData().isEmpty()) {
            logger.error("设备号{} 在线状态丢失", deviceId);
            return AjaxResult.error("客户端在线状态丢失");
        }
        MQTTClientStatus mqttClientStatus = respData.getData().get(0);
        if (mqttClientStatus == null || !mqttClientStatus.isConnected()) {
            logger.error("设备号{} 不在线", deviceId);
            return AjaxResult.error("客户端不在线");
        }
        ResponseEntity<String> subscriptionsResp = requestUtils.getSubscriptionsResp(mqttProperties.getApiUrl()  + "/api/v4/subscriptions/GID_cabinet@@@" + deviceId);
        if (subscriptionsResp.getStatusCodeValue() != 200) {
            logger.error("设备号{} 请求获取订阅接口错误", deviceId);
            return AjaxResult.error("获取订阅接口请求错误");
        }
        String emqxSubscriptionsResponseString = subscriptionsResp.getBody();
        if (StringUtils.isEmpty(emqxSubscriptionsResponseString)) {
            logger.error("设备号{} 请求获取订阅接口错误", deviceId);
            return AjaxResult.error("获取订阅接口请求错误");
        }

        logger.info("emqxSubscriptionsResponseString====" + emqxSubscriptionsResponseString);
        EmqxSubscriptionsResponse emqxSubscriptionsResponseBody = JSON.parseObject(emqxSubscriptionsResponseString, EmqxSubscriptionsResponse.class);
        if (emqxSubscriptionsResponseBody == null) {
            logger.error("设备号{} 请求获取订阅接口返回体错误", deviceId);
            return AjaxResult.error("获取订阅接口状态码错误");
        }
        int code = emqxSubscriptionsResponseBody.getCode();
        if (code != 0) {
            logger.error("设备号{} 请求获取订阅接口错误:{}", deviceId, emqxSubscriptionsResponseBody);
            return AjaxResult.error("获取订阅接口状态码错误");
        }
        List<EmqxSubscriptions> emqxSubscriptionsLists = emqxSubscriptionsResponseBody.getData();
        List<EmqxSubscriptions> collect = emqxSubscriptionsLists.stream().filter(s -> s.getTopic().equals("devices/" + deviceId)
        ).collect(Collectors.toList());
        if (collect.size() <= 0) {
            logger.error("设备号{} 设备未订阅该topic:{}", deviceId, emqxSubscriptionsLists);
            return AjaxResult.error("设备未订阅该topic");
        }
        Devices devices = devicesService.selectDevicesByDeviceId(deviceId);
        // 推送消息 使用client模式
        //emqClient.publish("devices/" + deviceId, data.toJSONString(), QosEnum.QoS1, false);
        // 使用http模式
//        String msgId = data.getString("msgId");
        String payload = buildMQTTMessage(event,devices.getRefreshToken(),biz);
        HashMap<String, Object> params = new HashMap<>();
        params.put("topic", "devices/" + deviceId);
        params.put("clientid", "GID_cabinet@@@" + deviceId);
        params.put("payload", payload);
        params.put("qos", 1);

        logger.info("MQTT发送日志====" + params);
        ResponseEntity<PublishRespEntity> responseEntity = requestUtils.postPublish(mqttProperties.getApiUrl()  + "/api/v4/mqtt/publish", params);
        if (responseEntity.getStatusCodeValue() != 200) {
            return AjaxResult.error("发消息状态错误," + responseEntity.getStatusCodeValue());
        }
        PublishRespEntity responseBody = responseEntity.getBody();
        if (responseBody == null) {
            logger.error("设备号{} 发消息返回体为空", deviceId);
            return AjaxResult.error("发消息返回体为空");
        }
        if (responseBody.getCode() != 0) {
            logger.error("设备号{} 发消息返回状态码错误:{}", deviceId, responseBody);
            return AjaxResult.error("发消息返回状态码错误," + responseBody);
        }
        boolean rrpc = body.getBooleanValue("rrpc");
        String orderId = body.getString("orderId");



        //log.info("rrpc:{},orderId:{},threadName:{}", rrpc, orderId,Thread.currentThread().getName());
        if (rrpc && !StringUtils.isEmpty(orderId)) {
            int count = 0;
            String rrpcKey = "RRPC:" + deviceId + ":" + orderId;
            long startTime2 = System.currentTimeMillis();
            while (true) {
                try {
                    Thread.sleep(1000);
                    boolean hasKey = redisCache.hasKey(rrpcKey);
                    if (hasKey) {
                        redisCache.del(rrpcKey);
                        logger.info("{} 收到rrpc回应, ThreadName:{}  count:{} 等待:{}ms", orderId, Thread.currentThread().getName(), count, (System.currentTimeMillis() - startTime2));
                        break;
                    }
                    if (count >= 10) {
                        redisCache.set(rrpcKey, 0);
                        redisCache.expire(rrpcKey, 60);
                        logger.error("RRPC:{} 等待客户端响应超时,请尝试再次扫码 ThreadName:{} ", orderId, Thread.currentThread().getName());
                        return AjaxResult.error("等待客户响应超时,请尝试再次扫码");
                    }
                    count++;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        logger.info("{} 发送消息 {} 到设备 topic:{},耗时{}ms ", mqttClientStatus.getClientid(), payload, emqxSubscriptionsLists, (System.currentTimeMillis() - startTime));
        return AjaxResult.success();
    }


    private String buildMQTTMessage(String event,String token,String biz) {

        JSONObject message = new JSONObject();
        message.put("uid","server");
        String messageId = DateUtil.date()+redisCache.getIncId(RedisConst.MESSAGE_ID_KEY,5);
        message.put("msgid",messageId);
        message.put("event",event);
        if (StringUtils.isNotEmpty(biz)) {
            message.put("biz",biz);
        }
        message.put("time",String.valueOf(Math.round(new Date().getTime()/1000)));
        message.put("apiv","1.0");
        message.put("sign", WXPayUtil.generateSignature(message,token));
        return message.toJSONString();
    }






}
